package com.usc.flink.datestream_api.state;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author jianjianDuan
 * @date 2020/7/16 9:01 下午
 *
 * 对数据去重，取每一个事件的第一条
 */
public class MyTestKeyedState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);

        DataStream<Tuple2<String, Integer>> ds = env.fromElements(
                new Tuple2<>("A", 2),
                new Tuple2<>("B", 3),
                new Tuple2<>("C", 4),
                new Tuple2<>("A", 5)
        );

        DataStream<Tuple2<String, Integer>> ds2 = ds.keyBy(e -> e.f0)
                .flatMap(new Duplicator());

        ds2.print();

        env.execute("Duplicate key");
    }

    private static class Duplicator extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {
        ValueState<Boolean> keyHasBeenSee;
        // 清除状态 keyHasBeenSeen.clear()
        @Override
        public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
            if (keyHasBeenSee.value() == null){
                out.collect(value);
                keyHasBeenSee.update(true);
            }
        }

        @Override
        public void open(Configuration conf) throws Exception {
            ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor("keyHasBeenSee", Types.BOOLEAN);
            keyHasBeenSee = this.getRuntimeContext().getState(desc);
        }
    }
}
